package com.cmnit.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cmnit.utils.ConfigurationManager;
import com.cmnit.utils.DateUtils;
import com.cmnit.utils.HttpUtils;
import com.cmnit.utils.MySqlUtils;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;

import java.io.IOException;
import java.util.*;

/**
 * 获取Yarn资源
 */
public class YarnService {
    private static YarnClient client = YarnClient.createYarnClient();

    // kerberos认证
    static {
        YarnConfiguration conf = new YarnConfiguration();
        conf.set("hadoop.security.authentication", "Kerberos");
        client.init(conf);
    }

    /**
     * 通过网页获取application对象信息
     */
    public static void getApplicationByXml() {
        String get = HttpUtils.doGet(ConfigurationManager.getProperty("http.application.xml"));
        // 获取全部application
        JSONObject jsonObject = JSON.parseObject(get);
        String appsBean = jsonObject.getString("apps");
        // 获取application数组
        jsonObject = JSON.parseObject(appsBean);
        String appArray = jsonObject.getString("app");
        // 获取数组对象
        JSONArray applications = JSON.parseArray(appArray);
        // 表名
        String tableName = "datalake_workspace_yarnapplication";
        // 需要写入的表字段
        String updateField = "applicationId,applicationName,startTime,queue,memorySize,virtualCores,time";

        // 获取指定字段
        for (Object application : applications) {
            JSONObject app = (JSONObject) application;
            if ("RUNNING".equals(app.getString("state"))) {
                // 获取任务ID
                String applicationId = app.getString("id");
                // 获取任务名
                String applicationName = app.getString("name");
                // 获取任务开始时间
                String startTime = DateUtils.getTimeByStamp(Long.valueOf(app.getString("startedTime")));
                // 获取任务队列
                String queue = app.getString("queue");
                // 获取任务内存大小
                String memorySize = app.getString("allocatedMB");
                // 获取任务CPU个数
                String virtualCores = app.getString("allocatedVCores");
                // 获取当前时间
                String time = DateUtils.getTimeString();
                // 写入字段的值
                String valuesField = applicationId + "," + applicationName + "," + startTime + "," + queue + "," + memorySize + "," + virtualCores + "," + time;
                MySqlUtils.insert(tableName, updateField, valuesField);
            }
        }
    }

    /**
     * 获取单个application对象的信息
     */
    public static void getYarnApplication() {
        client.start();

        // 获取指定状态的任务
        EnumSet<YarnApplicationState> appStates = EnumSet.noneOf(YarnApplicationState.class);
        if (appStates.isEmpty()) {
            appStates.add(YarnApplicationState.RUNNING);
        }

        // 返回EnumSet<YarnApplicationState>中个人任务状态的所有任务
        List<ApplicationReport> appsReport = null;
        try {
            appsReport = client.getApplications(appStates);
        } catch (YarnException | IOException e) {
            e.printStackTrace();
        }

        // 表名
        String tableName = "datalake_workspace_yarnapplication";
        // 需要写入的表字段
        String updateField = "applicationId,applicationName,startTime,queue,memorySize,virtualCores,time";

        // 获取所有的运行任务数据
        if (appsReport != null) {
            for (ApplicationReport appReport : appsReport) {
                // 获取任务ID
                String applicationId = appReport.getApplicationId().toString();
                // 获取任务名
                String applicationName = appReport.getName();
                // 获取任务开始时间
                String startTime = DateUtils.getTimeByStamp(appReport.getStartTime());
                // 获取任务队列
                String queue = appReport.getQueue();
                // 获取任务内存大小
                String memorySize = Long.toString(appReport.getApplicationResourceUsageReport().getUsedResources().getMemorySize());
                // 获取任务CPU个数
                String virtualCores = Integer.toString(appReport.getApplicationResourceUsageReport().getUsedResources().getVirtualCores());
                // 获取当前时间
                String time = DateUtils.getTimeString();

                // 写入字段的值
                String valuesField = applicationId + "," + applicationName + "," + startTime + "," + queue + "," + memorySize + "," + virtualCores + "," + time;
                MySqlUtils.insert(tableName, updateField, valuesField);
            }
        }
        client.stop();
    }


    /**
     * 获取yarn页面资源信息
     */
    public static void getYarnResources() {
        String get = HttpUtils.doGet(ConfigurationManager.getProperty("http.yarn.web"));
        JSONObject jsonObject = JSON.parseObject(get);
        // 获取yarn全部资源情况
        String bean = jsonObject.getString("beans");
        JSONArray objects = JSON.parseArray(bean);
        // 表名
        String tableName = "datalake_workspace_yarnstatics";
        // 需要写入的表字段
        String updateField = "AppsRunning,AllocatedMB,FairShareMB,AllocatedVCores,FairShareVCores,time";
        // 获取指定字段
        for (Object obj : objects) {
            JSONObject jsonOb = (JSONObject) obj;
            if ("Hadoop:service=ResourceManager,name=QueueMetrics,q0=root".equals(jsonOb.getString("name"))) {
                // 正在运行的APP个数
                String appsRunning = jsonOb.getString("AppsRunning");
                // 已分配内存数
                String allocatedMB = jsonOb.getString("AllocatedMB");
                // 总内存数
                String fairShareMB = jsonOb.getString("FairShareMB");
                // 已分配CPU个数
                String allocatedVCores = jsonOb.getString("AllocatedVCores");
                // 总CPU个数
                String fairShareVCores = jsonOb.getString("FairShareVCores");
                // 获取当前时间
                String time = DateUtils.getTimeString();
                // 写入字段的值
                String valuesField = appsRunning + "," + allocatedMB + "," + fairShareMB + "," + allocatedVCores + "," + fairShareVCores + "," + time;
                MySqlUtils.insert(tableName, updateField, valuesField);
            }
        }
    }
}
